1. Agent(智能体)的定义

LangChain 本质上是一个具备动态决策能力的智能执行框架，它通过大型语言模型（LLM）作为“大脑”，协调和调度各种工具来完成复杂任务。与传统的预定义流程软件不同，Agent 能够根据实际情况自主做出决策，将单一的工具调用转化为灵活的问题解决能力。

LLM Agent 在循环中运行工具以实现目标。代理运行直到满足停止条件-即，当模型发出最终输出或达到送代限制时停止迭代。LangChain 中 Agent 的架构由多个精密协作的组件构成，形成了一个完整的决策-执行-反馈循环系统。

Agent 的核心价值在于其三重核心能力：动态任务路由、生态化工具集成和全周期记忆管理。具体来说，动态任务路由使 Agent 能够根据输入内容自动规划执行路径，灵活切换工具调用逻辑；生态化工具集成让Agent 可以访问 300+ 预置工具接口，覆盖搜索引擎、数据计算、数据库交互等多领域；而全周期记忆管理则使 Agent 能同时维护短期对话上下文与长期知识存储，支持复杂任务的跨轮次协作。这种架构使得 Agent 不像传统程序那样按预定流程执行，而是更像一位智能项目经理，能够接收任务需求后拆解目标（Planmer 角色）、根招任务类型匹配最优工具（Router 角色）、调度工具按步骤执行（Executor 角色），最后整合结果生成交付物。

核心组件详解

• 模型（Model）：Model 是 Agent 的"大脑”，负责推理和决策过程。在 LangChain 1.0 中，Model 被抽象为统一的接口，可以与多种后端 LLM 提供商协同工作，包括 OpenAI、Anthropic、Google 等。Model 的核心职责是分析当前状态和用户请求，决定是否需要调用工具以及调用哪些工具，并解析工具返回结果以形成最终响应。Model 的推理过程通常基于==思维链（Chain-of-Thought）==模式，将复杂问题分解为多个思考步骤。
• 工具（Tools）：Tools 是 Agent 与外部世界交互的能力扩展接口，每个工具封装了一个特定功能，如搜索引擎查询、数据库操作、代码执行等。在 Langchan 中，工具即有大量预置选项（如搜索引擎即成、数学计算、文档检索等），也支持开发者自定义扩展。一个良好设计的工具应遵循功能单一性、输入验证和异常处理等原则。从架构角度看，工具将 Agent 的认知能力与具体执行能力分离，使得 Agent 可以专注于决策而非实现细节。
• 记忆（Memory）：Memory 为 Agent 提供上下文感知能力，使其能够记住之前的交互历史并基于上下文做出决策。LangChain 中的记忆系统分为短期记忆（维护当前对话的上下文）和长期记忆（支持跨对话会话的知识持久化）。在 LangChain 1.0 中，记忆系统与 LangGraph 的状态管理深度融合，支持更复杂和结构化的状态保持。
• AgentExecutor： 在 LangChain 1.0 的架构中，AgentExecutor 继续扮演执行协调器的关键角色，负责迭代运行代理直至满足停止条件。新版本中，AgentExecutor 基于 LangGraph 构建，获得了更强大的流程控制能力，包括循环控制（通过 max_iterations 参数防止无限循环）、异常处理（可配置的解析错误处理）和可观测性（通过 verbose 参数输出详细执行日志）

Agent 的工作流程与决策循环

具体流程如下：

• 输入解析：Agent 首先接收用户输入，解析其意图和关键参数。输入解析器负责标准化用户请求，提取可供工具使用的结构化参数。
• LLM 推理：解析后的输入与当前状态（包括记忆和历史记录）一起传递给 LLM 进行推理。LLM 基于预没的提示词横板分析情况，决定下一步行动——直接回答还是调用工具。
• 工具调用：如果 LLM 决定调用工具，AgentExecutor 会执行相应的工具函数，并获取执行结果。工具调用可能涉及外部 API 访问、数据库查询或计算操作等。
• 观察与迭代：工具返回的结果作为“观察“被反馈给 Agent，这些信息与之前的状态一起形成新的上下文，传递给 LLM 进行下一轮推理。这个循环持续进行，直到 LLM 认为已获得足够信息来生成最终答案。

Agent 是怎么工作的

![[Agent是怎么工作的.excalidraw|700]]
这个过程通常遵循一种名为 ==ReAct （Reason + Act）==的范式。以下是其工作流程：

步骤 1：接收输入与构建上下文
Agent 接收到用户的请求（例如：“上海今天天气怎么样？然后用摄氏度告诉我。“）。执行器会将此请求与之前的
对话历史（短期记忆）和任何相关长期记忆组合，形成当前的“上下文”。

步骤 2:LLM 推理与决策
执行器将这个完整的上下文输入给 LLM。LLM 会根据提示词的指引进行思考。提示词通常会要求模型以特定的格
式（如 JSON 或一段标记清晰的文本）输出它的“想法”和”下一步行动”。

• Reason（思考）：模型会先进行内部推理。
  “用户想知道上海的天气，并且要求用摄氏度表示。我自己没有实时天气数据，所以我需要一个工具来获取这些信息。可用的工具里有一个 Miget_weather 的工具，它可以查询天气。”

• Act（行动）：
  然后，模型会根据思考结果，决定一个行动。这个行动要么是最终回答，要么是调用一个工具。“我应该调用 get_weather 工具，参数是 location=上海。

2. LLM 模型組件

LangChain 支持所有主要模型提供商，包括 OpenAI、Anthropic、Google、Azure、AWS Bedrock 等。每个提供商都提供具有不同功能的各种模型

pip install -U "langchain［openai］" langchain-deepseek dotenv

一、各种供应商的大模型调用

国内外的各种大模型都可以直接使用 ChatOpenAI 类来负责调用。

在构建 ChatModels 时，我们有一些标准化参数：

• model：模型名称
• temperature：采样温度，值越高，大模型生成的答案越具有创造性
• timeout：请求超时
• max_tokens：生成的最大令牌数
• max_retries：请求重试的最大次数
• api_key：大模型供应商的 API 密钥
• base_url：发送请求的端点

from langchain_openai import ChatOpenAI
from langchain_deepseek import ChatDeepSeek

from env_utils import DEEPSEEK_API_KEY

# 无深度思考
llm = ChatOpenAI(
	model="deepseek-chat",
	temperature=0.5,
	base_url="https://api.deepseek.com",
	api_key=DEEPSEEK_API_KEY
)

llm = ChatDeepSeek(
	model="deepseek-chat",
	temperature=0.5,
	api_base="https://api.deepseek.com",
	api_key=DEEPSEEK_API_KEY
)

from test_deepseek import llm

resp = llm.invoke("用三句话介绍下机器学习的基本概念");
print(type(resp)) # <class 'langchain_core.messages.ai.AIMessage'>
print(resp)

二、流式输出

大多数模型都可以在生成输出内容时流式传输其输出内容。通过逐步显示输出，流媒体显着改善了用户体验，特别是对于较长的响应。stream() 与 invoke() 相反，invoke() 在模型生成完完整响应后返回单个 AIMessage，返回多个 AIMessageChunk 对象，每个对象包含输出文本的一部分。重要的是，流
中的每个块都设计为通过求和收集到完整消息中：

for chunk in llm.stream("用三句话介绍下机器学习的基本概念"):
	print(type(chunk)) # <class 'langchain_core.messages.ai.AIMessageChunk'>
	print(chunk.content) 
	# 深度思考
	reasoning_steps = [r for r in chunk.content_blocks if r['type'] == 'reasoning']
	print(reasoning_steps if reasoning_steps else chunk.text)

AIMessage 消息对象的属性

三、速率限制

许多聊天模型提供程序对给定时间段内可以进行的调用次数施加限制。如果达到速率限制，通常会收到来自提供商的速率限制错误响应，并且需要等待才能发出更多请求。为了帮助管理速率限制，聊天模型集成接受可在初始化期间提供的参数，以控制发出请求的速率。rate_limiter 参数来设置。

LangChain 内置 InMemoryRatelimiter。此限制器是线程安全的，可以由同一进程中的多个线程共享。

from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
  
rate_limiter = InMemoryRateLimiter(
	requests_per_second=0.1, # 每隔10s才能发送一个请求
	check_every_n_seconds=0.1, # 每100ms检查一次是否允许发出请求
	max_bucket_size=10, # 控制最大突发请求数量
)

# v1.0后才有的写法
model = init_chat_model(
	model="gpt-5",
	model_provider="openai",
	rate_limiter=rate_limiter,
)

四、模型的输出格式化

大型语言模型能够生成任意文本。这使得模型能够适当地响应广泛的输入范围，但对于某些用例，限制大型语言模型的输出为特定格式或结构是有用的。这被称为结构化输出。例如，如果输出要存储在关系数据库中，如果模型生成遵循定义的模式或格式的输出，将会容易得多。最常见的输出格式将是 JSON，尽管其他格式如 YAML 也可能很有用。

.with_structured_cutput()

为了方便，一些 LangChain 聊天模型支持，with_structured_output() 方法。该方法只需要一个模式作为输入，并返回一个字典或 Pydantic 对象。通常，这个方法仅在支持下面描述的更高级方法的模型上存在，并将在内部使用其中一种。它负责导入合适的输出解析器并将模式格式化为模型所需的正确格式。

from pydantic import BaseModel, Field

class Movie(BaseModel):
	title: str = Field(description="电影标题")
	year: int = Field(description="上映年份")
	director: str = Field(description="导演")
	rating: float = Field(description="评分") 

model_with_structure = llm.with_structured_output(Movie)
response = model_with_structure.invoke("介绍一部电影")
print(response) # title='肖申克的救赎' year=1994 director='弗兰克·德拉邦特' rating=9.3

SimpleJsop OutputParser

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import SimpleJsonOutputParser

prompt = ChatPromptTemplate.from_template(
	"尽你所能回答用户的问题" # 基本指令
	'你必须始终输出一个包含"title", "year", "director", "rating"的JSON对象，其中title是电影标题，year是上映年份，director是导演，rating是评分' # 输出结构
	"{question}" # 用户的问题
)

chain = prompt | llm | SimpleJsonOutputParser()
resp = chain.invoke({"question": "介绍一部电影"})
print(resp)

3. 创建 Agent 项目和本地测试环境

langchain 给我们提供了 Studio + LangSmith 集成的测试环境，使用 Studio 在本地可视化、交互和调试您的 Agent。

Studio 是一个专门的 Agent IDE，支持对实现智能体服务器 API 协议的代理系统进行可视化、交互和调试。Studio 还集成了跟踪、评估和提示工程。

一、安装 LangGraph CLI

LangGraph CLI 就是一个本地智能体的服务器环境

# Python >= 3.11 is required.
pip install --upgrade "langgraph-cli[inmem]"

二、配置 LangSmith 的环境变量

在项目的根目录中创建一个文件并填写必要的 API 密钥。我们需要将环境变量设置为从 LangSmith 获得的 API 密钥。.envLANGSMITH_API_KEY

LANGSMITH_API_KEY=Lsv2...

三、创建 LangGraph 配置文件

在应用程序的目录中，创建一个配置文件：langgraph.json

{
	"dependencies": ["."],
	"graphs": {
		"agent": "./src/agent.py:agent" // 智能体的文件位置
	},
	"env": ".env"
}

项目结构：

my-app/
|-—src
|  --agent.py
|--.env
|-- langgraph.json

四、编写你的智能体项目代码

from langchain.agents import create_agent

def send_email(to: str, subject: str, body: str):
	"""发送邮件到指定收件人。"""
	email = {	
		"to": to,
		"subject": subject,
		"body": body
	}
	# 邮件发送逻辑
	return f"邮件已发送至{to}"

agent = create_agent(
	"gpt-4o",
	tools=[send_email],
	system_prompt="你是一个邮件助手。请始终使用 send_email 工具。",
)

五、安装依赖项

在新 LangGraph 应用的根目录中，安装依赖项：

# 先拷贝pyproject.toml到项目目录下
pip install -e .

六、在 Studio 中启动和运行 agent

启动本地的 Agent 服务器：

langgraph dev

智能体程序将通过 Studio UI 访问: http://127.0.0.1:2024 https://smith.langchain.com/studio/?baseurl=http://127.0.0.1:2024

4. 工具的定义

工具是 Agent 调用以执行的组件。它们通过让模型通过定义明确的输入和输出与世界交互来扩展模型功能。

在构建 Agent 时，需要为其提供一个它可以使用的工具列表。除了实际调用的函数之外，工具还包括几个组件：

LangChain 支持从以下几种方式创建工具

1. Tool 装饰器的函数--这是最常见的。
2. 通过从 BaseTool 子类化-- 这是最灵活的方法，它提供了最大的控制程度，但代价是需要付出更多的努力和编写更多的代码。
3. 从 MCP 的服务端获得工具

一、@Tool 装饰器定义工具（简单工具）

from langchain_core.tools import tool  
from pydantic import BaseModel, Field  
  
from agent.my_llm import zhipuai_client  
  
# @tool('my_web_search', description="互联网搜索的工具，可以搜索所有公开的信息。", parse_docstring=True)  
@tool('my_web_search', parse_docstring=True)  
def web_search(query: str) -> str:  
    """互联网搜索的工具，可以搜索所有公开的信息。"""  # 普通的注释  
    """互联网搜索的工具，可以搜索所有公开的信息。  
  
    Args:        query: 需要进行互联网查询的信息  
            Returns:  
        返回搜索的结果信息，该信息是一个文本字符串。  
    """  # 谷歌的注释格式，可以定义工具的描述、参数、返回值  
  
    try:  
        resp = zhipuai_client.web_search.web_search(  
            search_engine='search_pro',  
            search_query=query,  
        )  
        if resp.search_result:  
            return "\n\n".join([d.content for d in resp.search_result])  
        return "没用搜索到任何结果"  
    except Exception as e:  
        print(e)  
        return f"Error: {e}"  
  
class SearchArgs(BaseModel):  
    query: str = Field(..., description="需要进行互联网查询的查询信息")  
  
@tool('my_web_search2', args_schema=SearchArgs, description="互联网搜索的工具，可以搜索所有公开的信息。",  
      parse_docstring=True)  
def web_search2(query: str) -> str:  
    pass  
  
# 本地测试  
if __name__ == '__main__':  
    print(web_search.name)  # 工具的名字  
    print(web_search.description)  # 工具的描述  
    print(web_search.args)  # 工具的参数  
    print(web_search.args_schema.model_json_schema())  # 工具的参数的json schema（描述json字符串）  
  
    result = web_search.invoke({'query': '如何使用langchain？'})  
    print(result)

from agent.tools.tool_demo1 import web_search  
from langchain.agents import create_agent  
from agent.my_llm import llm  
  
agent = create_agent(  
    llm,  
    tools=[web_search],  
    system_prompt="你是一个智能助手，尽可能的调用工具回答用户的问题。",  
)

二、组承 BaseTool 定义工具（复杂工具）

from typing import Type  
from langchain_core.tools import BaseTool  
from pydantic import BaseModel, Field, create_model  
from agent.my_llm import zhipuai_client  
  
class SearchArgs(BaseModel):  # 类：数据模型类  
    query: str = Field(..., description="需要进行互联网查询的查询信息")  
  
class MyWebSearchTool(BaseTool):  
    name: str = "web_search2"  # 定义工具的名称  
    description: str = "使用这个工具可以进行网络搜索"  # 定义工具的描述  
  
    # 第一种写法  
    # args_schema: Type[BaseModel] = SearchArgs  # 定义工具的参数  
    # 第二种写法  
    def __init__(self):  
        super().__init__()  
        self.args_schema = create_model("SearchInput",  
                                        query=(str, Field(..., description='需要进行互联网查询的查询信息')))  
  
    def _run(self, query: str) -> str:  
        try:  
            resp = zhipuai_client.web_search.web_search(  
                search_engine='search_pro',  
                search_query=query,  
            )  
            if resp.search_result:  
                return "\n\n".join([d.content for d in resp.search_result])  
            return "没用搜索到任何结果"  
        except Exception as e:  
            print(e)  
            return f"Error: {e}"
            
    # 定义异步工具  
	async def _run(self, query: str) -> str:  
    	return self._run(query)

from langchain.agents import create_agent  
from agent.my_llm import llm  
from agent.tools.tool_demo2 import MyWebSearchTool  
  
web_search_tool = MyWebSearchTool()  # 创建一个自定义的工具  
  
agent = create_agent(  
    llm,  
    tools=[web_search_tool],  
    system_prompt="你是一个智能助手，尽可能的调用工具回答用户的问题。",  
)

三、从 MGP 服务器端获得工具

后面的章节会讲

四、基于 Agent 的 Text-To-SQL 案例

使用人类的自然语言传给智能体，智能体生成 SQL 语句自动检查、执行 SQL 语句，并且将执行后的结果转换成人类理解的自然语言返回给用户

Agent 的工作原理

![[Agent 的工作原理.excalidaw]]

开发 Agent 的流桯：

1. 准备好数据库和配置数据连接，安装各种依赖库：sqlalchemy，pymysql，loguru
   1. sqlalchemy：操作关系型数据库的 ORM 框架库
   2. loguru：记录日志
2. 开发一个数据库操作的类：负责和数据库打交道。
3. 开发四个工具
4. 开发一个 Agent
5. 运行智能体

5. Agent Skills

Agent Skills = Skills 架构的 Agent。Skills 是模块化的能力，扩展了 Agent 的功能。每个 Skil 都打包了 LLM
指令、元数据、可选资源（脚本、模板等），Agent 会在需要时自动使用他们。

你可以把 Skills 理解为“通用 Agent 的扩展包”：Agent 可通过加载不同的 Skills 包，来具备不同的专业知识、工具使用
能力，稳定完成特定任务

Muti-Agent 架构和 Skills-Agent 架构

• Muti-Agent：多智能体架构
• Skills-Agent：技能列表架构

Muti-Agent 是什么

![[Muti-Agent 是什么.excalidraw|500]]

Agent Skills Multi-Agent 架构都旨在解决“如何让AI Agent处理复杂任务”的问题，但它们的思路、实现路径和适用场景有本质区别。

开发一个功能非常复杂的 Agent（能力很多）

开发任何一个 Agent 的思路：

1. 分析业务需求，捋清楚业务流程
2. 根据业务需求，评估复杂度来决定是否需要拆解（模块化），架构选择
3. 设计和开发可用的工具

两种不同的 Agent 开发架构：

1. 采用 Multi-Agent 架构
2. 采用 Agent-Skills 架构

6. Deep Agents

Deep Agents 是一个独立的库，可以利用文件系统，用于构建能够处理复杂、多步骤任务的智能体。Deep Agents 基于 LangGraph 构建，并受到 Claude Code、Deep Research 和 Manus 等应用程序的启发，具有规划能力、用于上下文管理的文件系统以及生成子 agent 的能力。

class SubAgent （TypedDict）：
	name: str
	description: str
	prompt: str
	tools: Sequence［BaseTool | Callable | ut［str, Any］］
	model: NotRequired［str | BaseChatModel］
	middleware: NotRequired［list［AgentMiddleware］］
	interrupt_on: NotRequired［dict［str, bool | InterruptOnConfig］］
	
class CompiledSubAgent （TypedDict）：
	name: str
	description: str
	runnable: Runnable

![[Deep Agents.excalidraw]]